package org.apache.rocketmq.store;

import org.apache.rocketmq.common.ServiceThread;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;

import java.util.concurrent.LinkedBlockingQueue;
import org.apache.rocketmq.store.CommitLog.GroupCommitRequest;

/* @方法描述：执行同步刷新数据操作时要用到的硬盘状态监听器，防止同步等待刷新数据操作完成的线程一直等待结果
 * 如果刷新操作超时，这个硬盘监听器就会直接唤醒同步等待刷新数据操作完成的线程
 */
public class FlushDiskWatcher extends ServiceThread {
    private static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);

    //同步刷盘请求的提交队列
    private final LinkedBlockingQueue<CommitLog.GroupCommitRequest> commitRequests = new LinkedBlockingQueue<>();
    @Override
    public String getServiceName() {
        return FlushDiskWatcher.class.getSimpleName();
    }

    @Override
    public void run() {
        while (!this.isStopped()){
           GroupCommitRequest request = null;
           try {
               request = commitRequests.take();
           } catch (InterruptedException e) {
               log.error("FlushDiskWatcher take request error", e);
               continue;
           }
           //直到刷盘完成
            while(!request.future().isDone()){
                //只有刷盘完成时才退出循环
                long now =System.nanoTime();
                //判断是否已经超过了同步刷盘的时间
                if (now - request.getDeadLine() >= 0 ){
                    request.wakeupCustomer(PutMessageStatus.FLUSH_DISK_TIMEOUT);
                }
                //计算当前线程睡眠时间
                long sleepTime = (request.getDeadLine() - now) / 1_000_000;
                //限制当前线程最大睡眠时间为10ms
                sleepTime = Math.min(10, sleepTime);
                //如果计算的当前线程睡眠时间为0，这就意味着equest.getDeadLine() = now了
                //操作已经超时了
                if (sleepTime == 0) {
                    //直接唤醒等待同步操作完成的线程
                    request.wakeupCustomer(PutMessageStatus.FLUSH_DISK_TIMEOUT);
                    break;
                }

                try {
                    //让当前线程睡一会，然后进入下一个循环
                    Thread.sleep(sleepTime);
                } catch (InterruptedException e) {
                    log.warn("An exception occurred while waiting for flushing disk to complete. this may caused by shutdown");
                    break;
                }
            }
        }
    }

    public void add(GroupCommitRequest request) {
        commitRequests.add(request);
    }

    public int queueSize() {
        return commitRequests.size();
    }
}